package flink_p2_sql

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.{Slide, Table, TableEnvironment, Tumble}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.types.Row


object tableApi05_tableApi_window {


  /**
   * @param args
   */
  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)


    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.scala._


    val dataStream: DataStream[(Long, String, Int)] = env.socketTextStream("node1", 8889)
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(3)) {
        override def extractTimestamp(element: String): Long = {
          element.split(" ")(0).toLong
        }
      })
      .map(data => {
      val arrs: Array[String] = data.split(" ")
      (arrs(0).toLong, arrs(1), arrs(2).toInt)
    })



    // 将DataStream 转换为 Table
    val table: Table = tableEnv.fromDataStream(dataStream, 'time.rowtime, 'name, 'age)

    table.printSchema()

    /**
     * 滚动窗口：每5秒统计最近5秒 ，每个年龄的人数
     */


//     Tumble.over("5.second") => 5秒的滚动窗口
    val resTable: Table = table.window(Tumble.over("5.second").on("time").as("win"))
      .groupBy('win, 'age)
      .select('age, 'win.start, 'win.end, 'win.rowtime, 'age.count)


    tableEnv.toRetractStream[Row](resTable).print()


    /**
     * 滑动窗口：每5秒统计一次最近10秒之内每个年龄的人数
     */
//    val res1: Table = table.window(Slide.over("5.second").every("10.second").on("time").as("win"))
//      .groupBy('win, 'age)
//      .select('age, 'age.count)

//    res1.toRetractStream[Row].print()

    tableEnv.execute("table api window")
  }
}
